package com.cssl.app.function;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;

/**
 * @Author: chen
 * @Date: 2021/11/10 21:26
 * @Desc: 自定义 flink cdc 反序列化器
 */
public class CustomerDeserialization implements DebeziumDeserializationSchema<String> {

    /**
     * 封装数据格式 json
     * {
     * "database":"库名",
     * "tableName":"表名",
     * "type":"操作类型",
     * "before":{"字段1":"","字段2":"","字段3":""},
     * "after":{"字段1":"","字段2":"","字段3":""}
     * }
     */
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
        //1. 创建json对象用于存储最终数据
        JSONObject result = new JSONObject();

        //2. 获取库名和表名
        String topic = sourceRecord.topic();
        String[] fields = topic.split("\\.");
        result.put("database", fields[1]);
        result.put("tableName", fields[2]);

        Struct value = (Struct) sourceRecord.value();
        //3. 获取before数据
        Struct before = value.getStruct("before");
        JSONObject beforeJson = new JSONObject();
        if (before != null) {
            Schema beforeSchema = before.schema();
            List<Field> beforeFields = beforeSchema.fields();
            for (Field field : beforeFields) {
                Object beforeValue = before.get(field);
                beforeJson.put(field.name(), beforeValue);
            }
        }
        result.put("before", beforeJson);

        //4. 获取after数据
        Struct after = value.getStruct("after");
        JSONObject afterJson = new JSONObject();
        if (after != null) {
            Schema afterSchema = after.schema();
            List<Field> afterFields = afterSchema.fields();
            for (Field field : afterFields) {
                Object afterValue = after.get(field);
                afterJson.put(field.name(), afterValue);
            }
        }
        result.put("after", afterJson);

        //5. 获取操作类型 insert update delete
        Envelope.Operation operation = Envelope.operationFor(sourceRecord);
        String type = operation.toString().toLowerCase();
        if ("create".equals(type)) {
            type = "insert";
        }
        result.put("type", type);

        //6. 输出
        collector.collect(result.toJSONString());
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }
}
